package com.pk.flink.scenario01.app;

import com.pk.flink.scenario01.bean.NotFoundElement;
import com.pk.flink.scenario01.sink.NotFoundElementRedisSink;
import com.pk.flink.scenario01.source.NotFoundElementSource;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;

import java.text.SimpleDateFormat;

public class NotFoundElementSumApp {
    public static void main(String[] args) throws Exception {
        Configuration config = new Configuration();
        // 网络缓冲区占TaskManager内存的比例（默认0.1，调高至0.3）
        config.setFloat("taskmanager.memory.network.fraction", 0.3f);
        // 网络缓冲区最小内存（默认64MB，可设为128MB）
        config.setString("taskmanager.memory.network.min", "256mb");
        // 网络缓冲区最大内存（默认1GB，可设为256MB）
        config.setString("taskmanager.memory.network.max", "256mb");

        //获取上下文环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd-HH");
        SimpleDateFormat sdfFull = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

        DataStreamSource<NotFoundElement> ntfDataStreamSource = env.addSource(new NotFoundElementSource()).setParallelism(1);
        SingleOutputStreamOperator<Tuple3<String, String, String>> sum =
                ntfDataStreamSource.map(x -> Tuple3.of(x.getWebpageResource(), sdf.format(sdfFull.parse(x.getVisitTime())), 1))
                .returns(Types.TUPLE(Types.STRING, Types.STRING, Types.INT))
                .keyBy(new KeySelector<Tuple3<String, String, Integer>, Tuple2<String, String>>() {
                    @Override
                    public Tuple2<String, String> getKey(Tuple3<String, String, Integer> value) throws Exception {
                        return Tuple2.of(value.f0, value.f1);
                    }
                })
                .sum(2)
                .map(x -> Tuple3.of("pk_product_" + x.f1, x.f0, x.f2.toString()))
                        .returns(Types.TUPLE(Types.STRING, Types.STRING, Types.STRING));
//        sum.print();
        sum.addSink(new NotFoundElementRedisSink());
        env.execute();
    }
}
